
package com.showdor.springboot;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

/**
 * @Author: lijb6
 * @Version 1.0
 */
//测试组消费
//@Component
public class GroupConsumer {
    private final Logger logger = LoggerFactory.getLogger(GroupConsumer.class);
    private static int count = 0;

    //组1，消费者1
    @KafkaListener(containerGroup = "sync_callback_topic", topicPartitions = {@TopicPartition(topic = "netcross", partitions = {"0"})})
    public void onMessage1(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            count += 1;
            logger.info("{}-获得数据:{}", count, String.valueOf(msg));
            //WriterDataFile.writeData(count + "," + JSONObject.toJSONString(msg));
        }
    }

    //组1，消费者2
    @KafkaListener(topics = {"netcross_violation"}, groupId = "group1",containerFactory = "kafkaListenerContainerFactory2")
    public void onMessage2(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        logger.info("本次消费量:{}",records.size());
        records.forEach(record -> {
            logger.info("数据：{}",record.value());
            // 处理每条消息
        });
        ack.acknowledge();
    }

    //组2，只有一个消费者
    @KafkaListener(topics = {"test"}, groupId = "group2")
    public void onMessage3(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            logger.info("group:group2 , message:{}", msg);
        }
    }
}

